%%% -------------------------------------------------------------------
%%% Author  : vjache
%%% Description : Ring Route Service.
%%%
%%% Created : 19.02.2009
%%% -------------------------------------------------------------------
-module(svc_rrs).

-behaviour(gen_server).
%% --------------------------------------------------------------------
%% Include files
%% --------------------------------------------------------------------

-include("../include/rrs.hrl").

-define(SERVER,?MODULE).
-define(HASH_MAX,10000000).

%% --------------------------------------------------------------------
%% External exports
-export([
		 start_link/1,
		 get_nodes/0,
		 get_nodes/1,
		 make_route_snapshot/0,
		 key_route/2,
		 key_route/1,
		 pipe_router/2,
		 pipe_router/3,
		 hash/1,
		 resp_stats/0]).

%% gen_server callbacks
-export([
		 init/1, 
		 handle_call/3, 
		 handle_cast/2, 
		 handle_info/2, 
		 terminate/2, 
		 code_change/3]).

-record(state, {}).

-record(?RING_TABLE, {point,node}).

%% ====================================================================
%% External functions
%% ====================================================================
key_route(Key,RouteTable) ->
	Gbt={_Size,Tree}=RouteTable,
	KeyHash=hash(Key),
	case gbt_ring_seek(KeyHash,Tree,smallest) of
		smallest -> {_K,V}=gb_trees:smallest(Gbt),V;
		Route -> Route
	end.

key_route(Key) ->
	key_route(Key,make_route_snapshot()).


hash(Term) ->
	erlang:phash2({"976",Term},?HASH_MAX).

make_route_snapshot() ->
	gen_server:call(?SERVER, mk_route_snapshot).

get_nodes() ->
	Snap=make_route_snapshot(),
	get_nodes(Snap).

get_nodes(Snap) ->
	{Up,Down}=lists:foldl(
				fun({ok,UpNode},{UpSet,DownSet}) ->
						{sets:add_element(UpNode, UpSet),DownSet};
				   ({redirect,DownNode,_UpNode},{UpSet,DownSet}) ->
						{UpSet,sets:add_element(DownNode, DownSet)}
				end
				,{sets:new(),sets:new()},gb_trees:values(Snap)),
	{sets:to_list(Up),sets:to_list(Down)}.

gbt_ring_seek(KeyHash,{K,_V,_Sm,Bg}=_Tree,Res) when KeyHash>K ->
	gbt_ring_seek(KeyHash,Bg,Res);
gbt_ring_seek(KeyHash,{K,V,Sm,_Bg}=_Tree,_Res) when KeyHash<K ->
	gbt_ring_seek(KeyHash,Sm,V);
gbt_ring_seek(KeyHash,{K,_V,_Sm,_Bg}=_Tree,Res) when KeyHash==K ->
	Res;
gbt_ring_seek(_KeyHash,nil,Res) ->
	Res.

resp_stats() ->
	Gbt=make_route_snapshot(),
	RouteList=gb_trees:to_list(Gbt),
	{Crp,Dict}=lists:foldl(fun({Rp,Route},{Crp,Dict}) ->
								   Node=case Route of
											{ok,UpNode} -> UpNode;
											{redirect,DownNode,_UpNode} -> DownNode
										end,
								   Delta=Rp-Crp,
								   {Rp,dict:update_counter(Node, Delta, Dict)}
						   end, {0,dict:new()}, RouteList),
	[{_Rp,FirstRoute}|_]=RouteList,
	FirstNode = route2node(FirstRoute),
	Dict1=dict:update_counter(FirstNode, ?HASH_MAX-Crp, Dict),
	[{N,R/?HASH_MAX}||{N,R}<-dict:to_list(Dict1)].

pipe_router(KeyFun,NextPipe) ->
	pipe_router(KeyFun,make_route_snapshot(),NextPipe).
pipe_router(KeyFun,RouteSnap,NextPipe) ->
	pipe_mapper:new(fun(Item)->
							Key=KeyFun(Item),
							Node=?ROUTE2UPNODE(key_route(Key, RouteSnap)),
							{Node,Item}
					end, NextPipe).

%% ====================================================================
%% Server functions
%% ====================================================================
start_link(Options) ->
	io:format("svc_rrs:start_link~n"),
    gen_server:start_link({local,?SERVER}, ?MODULE, _InitArgs=[], Options).

%% --------------------------------------------------------------------
%% Function: init/1
%% Description: Initiates the server
%% Returns: {ok, State}          |
%%          {ok, State, Timeout} |
%%          ignore               |
%%          {stop, Reason}
%% --------------------------------------------------------------------
init([]) ->
	io:format("svc_rrs:init~n"),
	init_mnesia(svc_cms:get_randezvouse()),
	init_ring_info(),
	io:format("svc_rrs:init 1~n"),
    {ok, #state{}}.

%% --------------------------------------------------------------------
%% Function: handle_call/3
%% Description: Handling call messages
%% Returns: {reply, Reply, State}          |
%%          {reply, Reply, State, Timeout} |
%%          {noreply, State}               |
%%          {noreply, State, Timeout}      |
%%          {stop, Reason, Reply, State}   | (terminate/2 is called)
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_call(mk_route_snapshot, _From, State) ->
	Reply = mk_route_snapshot_impl(),
    {reply, Reply, State};
handle_call(_Request, _From, State) ->
    Reply = ok,
    {reply, Reply, State}.

%% --------------------------------------------------------------------
%% Function: handle_cast/2
%% Description: Handling cast messages
%% Returns: {noreply, State}          |
%%          {noreply, State, Timeout} |
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_cast(_Msg, State) ->
    {noreply, State}.

%% --------------------------------------------------------------------
%% Function: handle_info/2
%% Description: Handling all non call/cast messages
%% Returns: {noreply, State}          |
%%          {noreply, State, Timeout} |
%%          {stop, Reason, State}            (terminate/2 is called)
%% --------------------------------------------------------------------
handle_info(_Info, State) ->
    {noreply, State}.

%% --------------------------------------------------------------------
%% Function: terminate/2
%% Description: Shutdown the server
%% Returns: any (ignored by gen_server)
%% --------------------------------------------------------------------
terminate(_Reason, _State) ->
    ok.

%% --------------------------------------------------------------------
%% Func: code_change/3
%% Purpose: Convert process state when code is changed
%% Returns: {ok, NewState}
%% --------------------------------------------------------------------
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%% --------------------------------------------------------------------
%%% Internal functions
%% --------------------------------------------------------------------
init_mnesia(RandNode)->
	case node() of
		nonode@nohost -> exit(node_is_not_alive);
		RandNode ->
			case mnesia:create_table(?RING_TABLE,
									 [
									  {disc_copies, [RandNode]},
									  {index,[#?RING_TABLE.node]},
									  {type, ordered_set},
									  {attributes, record_info(fields, ?RING_TABLE)}]) of
				{atomic,ok} -> ok;
				{aborted,{already_exists,?RING_TABLE}}->ok;
				Err -> exit(Err)
			end;
		Node ->
			case mnesia:add_table_copy(?RING_TABLE, Node, disc_copies) of
				{aborted,{already_exists,?RING_TABLE,_}} -> ok;
				{atomic,ok} -> ok
			end
	end,
	ok=wait_for_tables(start,[?RING_TABLE]).

wait_for_tables(ok,TabList) ->
	io:format("Tables ~p are ready.~n",[TabList]),ok;
wait_for_tables({timeout,TabList},TabList) ->
	io:format("Tables ~p still not ready...~n",[TabList]),
	wait_for_tables(start,TabList);
wait_for_tables({error,_Reason}=Err,_TabList) -> exit(Err);
wait_for_tables(start,TabList) ->
	io:format("Wait tables ~p become ready...~n",[TabList]),
	wait_for_tables(mnesia:wait_for_tables(TabList, 1*1000),TabList).

init_ring_info() ->
	mnesia:sync_transaction(
	  fun()->
			  Node=node(),
			  NodeAsStr=atom_to_list(Node),
			  CList=lists:seq(1, 100),
			  lists:foreach(fun(C)->
									RingEntry=#?RING_TABLE{point=hash([C,NodeAsStr]),node=Node},
									case mnesia:read({?RING_TABLE,RingEntry#?RING_TABLE.point}) of
										[RingEntry]->ok;
										[]->mnesia:write(RingEntry);
										ConflictRingEntry -> exit({ring_init_coflict,{in_db,ConflictRingEntry},{tried,RingEntry}})
									end
							end, CList)
	  end).

mk_route_snapshot_impl() ->
	NodesUp=lists:foldl(fun(Node,SetAcc)->
								gb_sets:insert(Node, SetAcc)
						end,gb_sets:new(),mnesia:system_info(running_db_nodes)),
	UpFun=fun(Node)-> 
				  case gb_sets:is_member(Node, NodesUp) of
					  true -> up;
					  false -> down
				  end
		  end,
	FoldFun=fun({P,{down,Node}},{C,Result})->
					{C,gb_trees:enter(P, {redirect,Node,C}, Result)};
			   ({P,{up,Node}},{_C,Result})->
					{Node,gb_trees:enter(P,{ok,Node},Result)}
			end,
	{atomic,{_FirstUp,Rs}}=mnesia:transaction(
							 fun() ->
									 mnesia:foldr(
									   fun(#?RING_TABLE{point=Point,node=Node},Acc) ->
											   FoldFun({Point, {UpFun(Node),Node}},Acc)
									   end, {first_up(UpFun),gb_trees:empty()}, ?RING_TABLE) 
							 end),
	Rs.

route2node(Route) ->
	case Route of
		{ok,UpNode} -> UpNode;
		{redirect,DownNode,_UpNode} -> DownNode
	end.

first_up(UpFun)->
	first_up_loop(mnesia:first(?RING_TABLE),UpFun).
first_up_loop('$end_of_table',_UpFun)->
	exit({ring_empty});
first_up_loop(Key,UpFun)->
	[#?RING_TABLE{point=Key,node=Node}]=mnesia:read({?RING_TABLE, Key}),
	case UpFun(Node) of
		up -> Node;
		down -> first_up_loop(mnesia:next(?RING_TABLE, Key),UpFun)
	end.